package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.DwsScoreDistributedWindonBean;
import com.atguigu.edu.realtime.bean.DwsTestExamWindonBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsScoreDistributedWindon {

    public static void main(String[] args) throws Exception {
        // TODO 1、获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        // TODO  3、设置检查点和状态后端
        /*env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        //延迟时间
        env.getCheckpointConfig().setCheckpointTimeout(3 * 60 * 1000L);
        //最多可以出现多少状态后端
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //检查点保存位置
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //开启状态后端
        env.setStateBackend(new HashMapStateBackend());*/

        //TODO 4、读取kafka数据
        String topicName = "dwd_text_exam";
        String groupId = "dws_score_distributed_windon";

        DataStreamSource<String> kakfaStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));

        //TODO 5、过滤转换数据

        SingleOutputStreamOperator<DwsScoreDistributedWindonBean> flatMapStream = kakfaStream.map(JSONObject::parseObject).keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("paper_id") + value.getString("user_id");
            }
        }).flatMap(new RichFlatMapFunction<JSONObject, DwsScoreDistributedWindonBean>() {
            ValueState<String> pid_uid = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                pid_uid = getRuntimeContext().getState(new ValueStateDescriptor<String>("pid_uid", String.class));
            }

            @Override
            public void flatMap(JSONObject value, Collector<DwsScoreDistributedWindonBean> out) throws Exception {

                String statevalue1 = pid_uid.value();
                String key = value.getString("paper_id") + value.getString("user_id");
                String score_segmenta = null;
                if (value.getLong("score") > 0 && value.getLong("score") < 60) {
                    score_segmenta = "(0-60)";
                } else if (value.getLong("score") >= 60 && value.getLong("score") < 80) {
                    score_segmenta = "[60-80)";
                } else if (value.getLong("score") >= 80 && value.getLong("score") <= 100) {
                    score_segmenta = "[60-100]";
                } else {
                    score_segmenta = "脏数据";
                }

                if (!key.equals(statevalue1)) {
                    out.collect(DwsScoreDistributedWindonBean.builder()
                            .paper_id(value.getString("paper_id"))
                            .score_segmenta(score_segmenta)
                            .user_cnt(1L)
                            .ts(DateFormatUtil.toTs(value.getString("submit_time"), true))
                            .build());
                    pid_uid.update(value.getString("user_id"));
                } else {
                }

            }
        });
        //TODO 6、添加水位线开窗
        SingleOutputStreamOperator<DwsScoreDistributedWindonBean> watermarkStraem = flatMapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<DwsScoreDistributedWindonBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<DwsScoreDistributedWindonBean>() {
            @Override
            public long extractTimestamp(DwsScoreDistributedWindonBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        KeyedStream<DwsScoreDistributedWindonBean, String> keyedStream = watermarkStraem.keyBy(new KeySelector<DwsScoreDistributedWindonBean, String>() {
            @Override
            public String getKey(DwsScoreDistributedWindonBean value) throws Exception {
                return value.getPaper_id()+value.getScore_segmenta();
            }
        });
        SingleOutputStreamOperator<DwsScoreDistributedWindonBean> reduceStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)))
                .reduce(new ReduceFunction<DwsScoreDistributedWindonBean>() {
                    @Override
                    public DwsScoreDistributedWindonBean reduce(DwsScoreDistributedWindonBean value1, DwsScoreDistributedWindonBean value2) throws Exception {
                        value1.setUser_cnt(value1.getUser_cnt() + value2.getUser_cnt());
                        return value1;
                    }
                }, new WindowFunction<DwsScoreDistributedWindonBean, DwsScoreDistributedWindonBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<DwsScoreDistributedWindonBean> input, Collector<DwsScoreDistributedWindonBean> out) throws Exception {
                        DwsScoreDistributedWindonBean dwsScoreDistributedWindonBean = input.iterator().next();
                        dwsScoreDistributedWindonBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        dwsScoreDistributedWindonBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        dwsScoreDistributedWindonBean.setSystime(System.currentTimeMillis());
                        out.collect(dwsScoreDistributedWindonBean);
                    }
                });

        reduceStream.print("sssss");


        env.execute(groupId);

    }
}
